package cn.akwangl.config;

import cn.akwangl.interceptor.LoggerWrapper;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ttl.threadpool.TtlExecutors;
import org.slf4j.Logger;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.scheduling.annotation.*;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.function.SingletonSupplier;

import java.util.Collection;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 异步线程池配置.
 * <p>
 * 如果不实现该类则会使用内置的线程池{@link SimpleAsyncTaskExecutor},这个线程池每添加一个task就会创建一个线程,频繁创建销毁线程性能开销非常大.
 * 该类会{@link AbstractAsyncConfiguration#setConfigurers(Collection)}方法中加载,这是一个懒加载方式,
 * 并在{@link ProxyAsyncConfiguration#asyncAdvisor}设置到{@link AsyncAnnotationBeanPostProcessor#executor}中.
 * <pre>
 *     	AsyncAnnotationBeanPostProcessor bpp = new AsyncAnnotationBeanPostProcessor();
 *     	// 设置线程池、异常处理
 * 		bpp.configure(this.executor, this.exceptionHandler);
 * </pre>
 * 假如{@link AsyncAnnotationBeanPostProcessor#executor}为NULL,则使用默认的线程池,获取线程池在{@link SingletonSupplier#get()}方法中有所体现.
 * {@code    this.defaultExecutor = new SingletonSupplier<>(defaultExecutor, () -> getDefaultExecutor(this.beanFactory));}
 * <p>
 * 参考文献:<br/>
 * <a href="https://blog.csdn.net/yaomingyang/article/details/108165496">springboot通过AsyncConfigurer接口实现异步线程池自动化配置组件</a><br/>
 * <a href="https://xie.infoq.cn/article/2921fdbae94fee573e8cf9b91">Spring 源码学习 - @Async 注解实现原理</a><br/>
 *
 * @author 阿坤同学
 * @date 2022/5/26 16:33
 * @see SingletonSupplier
 * @since 0.0.4
 */
@EnableAsync
@Configuration
public class AsyncScheduledTaskConfig implements AsyncConfigurer {

    private static final Logger log = LoggerWrapper.getLogger(AsyncScheduledTaskConfig.class);

    /**
     * W:工作时间
     * C:休息时间
     * Ncpu:线程核心数
     * Ucpu: CPU利用率
     * <p>
     * CPU 型: (1+W/C) * Ncpu * Ucpu(cpu利用率)
     * IO 型: (W+C)/C * Ncpu * Ucpu(cpu利用率)
     */
    private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() / 2;
    private static final int MAX_POOL_SIZE = (int) ((CORE_POOL_SIZE * 2) * 0.8);
    /**
     * 允许线程空闲时间（单位为秒）
     */
    private static final int KEEP_ALIVE_TIME = 10;
    /**
     * 缓冲队列数
     */
    private static final int QUEUE_CAPACITY = 200;
    /**
     * 线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁
     */
    private static final int AWAIT_TERMINATION = 60;
    /**
     * 用来设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
     */
    private static final Boolean WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN = true;
    /**
     * 线程池名前缀
     */
    private static final String THREAD_NAME_PREFIX = "Async-Service-";

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(CORE_POOL_SIZE);
        taskExecutor.setMaxPoolSize(MAX_POOL_SIZE);
        taskExecutor.setKeepAliveSeconds(KEEP_ALIVE_TIME);
        taskExecutor.setQueueCapacity(QUEUE_CAPACITY);
        taskExecutor.setThreadNamePrefix(THREAD_NAME_PREFIX);
        taskExecutor.setWaitForTasksToCompleteOnShutdown(WAIT_FOR_TASKS_TO_COMPLETE_ON_SHUTDOWN);
        taskExecutor.setAwaitTerminationSeconds(AWAIT_TERMINATION);
        // 线程池对拒绝任务的处理策略
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 初始化
        taskExecutor.initialize();
        return TtlExecutors.getTtlExecutor(taskExecutor);
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        AsyncUncaughtExceptionHandler asyncUncaughtExceptionHandler = (ex, method, params) -> {
            String paramsJson = null;
            if (params != null && params.length > 0) {
                paramsJson = JSONObject.toJSONString(params);
            }
            String msg = String.format("异步任务执行错误,执行的方法:%s,方法参数个数:%s,方法参数:%s", method.getName(), method.getParameterCount(), paramsJson);
            log.error(msg, ex);
        };
        return asyncUncaughtExceptionHandler;
    }
}